fix(inkless): repair missing control-plane rows on CreatePartitions retry [KC-420] by viktorsomogyi · Pull Request #752 · aiven/inkless

viktorsomogyi · 2026-08-14T13:21:04Z

A failed control-plane write after KRaft already applied the increase left partitions advertised but unusable; retry returned INVALID_PARTITIONS and skipped the insert. Also treat a diskless LATEST below the committed seal as the seal, so a placeholder row cannot truncate the classic prefix.

Copilot

Pull request overview

This PR hardens Inkless diskless-topic partition creation and consolidation behavior by making CreatePartitions retries repair missing control-plane rows and by preventing placeholder control-plane LATEST offsets from incorrectly truncating classic prefixes below the committed KRaft seal.

Changes:

Update controller CreatePartitions handling to generate control-plane insert requests for “born-diskless” partitions on retries, including non-contiguous ranges and skipping sealed/switching partitions when the image indicates them.
Clamp diskless LATEST offsets (and epoch end offsets derived from LATEST) to at least the committed seal to avoid truncation from placeholder rows.
Expand unit test coverage for both behaviors across controller and diskless leader endpoint paths.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
storage/inkless/src/main/java/io/aiven/inkless/control_plane/CreateTopicAndPartitionsRequest.java	Updates API documentation to reflect range-based partition insertion semantics for retries/holes.
core/src/main/scala/kafka/server/ControllerApis.scala	Implements retry-aware control-plane row insertion, including partition classification and contiguous range batching.
core/src/test/scala/unit/kafka/server/ControllerApisTest.scala	Adds tests covering CreatePartitions retry scenarios, sealed/switching skips, non-contiguous insertion, and decrease rejection.
core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala	Ensures LATEST-derived offsets are never below the committed seal and keeps leader epoch resolution consistent with the adjusted offset.
core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala	Adds tests validating seal clamping behavior for LATEST and for epoch end offsets.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

…etry A failed control-plane write after KRaft already applied the increase left partitions advertised but unusable; retry returned INVALID_PARTITIONS and skipped the insert. Also treat a diskless LATEST below the committed seal as the seal, so a placeholder row cannot truncate the classic prefix. Co-authored-by: Cursor <cursoragent@cursor.com>

jeqo

Looking good. Left a few comments before approving.

jeqo · 2026-08-17T18:16:44Z

+    verify(controlPlane).createTopicAndPartitions(ArgumentMatchers.eq(singleton(
+      new CreateTopicAndPartitionsRequest(topicId, topicName, 3, 4))))
+  }
+


for completeness, we can add the scenario that should still fail and do not insert rows:

Suggested change

@Test

def testCreateDisklessPartitionsDecreaseRetryDoesNotInsert(): Unit = {

// topic image already has 4 partitions; a decrease to 2 is rejected by KRaft with

// INVALID_PARTITIONS and must NOT write control-plane rows.

val topic = new CreatePartitionsTopic().setName(topicName).setAssignments(null).setCount(2)

val result = new CreatePartitionsTopicResult().setName(topicName).setErrorCode(INVALID_PARTITIONS.code())

// ... arrange metadataCache image with 4 partitions for topicId (as the other retry tests do) ...

// invoke createDisklessPartitions(...)

verify(controlPlane, never()).createTopicAndPartitions(any())

}

jeqo · 2026-08-17T18:23:51Z

+    val retry = errorCode != Errors.NONE.code()
+    if (retry && count < imagePartitionCount(topicId)) {
+      Seq.empty
+    } else {
+      val firstPartition =
+        if (retry) 0
+        else priorTopicStates.get(topicName) match {
+          case Some(state) if state.topicId == topicId => math.min(state.numPartitions, count)
+          case _ => 0
+        }
+      val partitions = bornDisklessPartitions(topicId, count).filter(_ >= firstPartition)


imagePartitionCount and bornDisklessPartitions both access metadataCache.currentImage() snapshot at different points, potentially getting different images; maybe:

Suggested change

val retry = errorCode != Errors.NONE.code()

if (retry && count < imagePartitionCount(topicId)) {

Seq.empty

} else {

val firstPartition =

if (retry) 0

else priorTopicStates.get(topicName) match {

case Some(state) if state.topicId == topicId => math.min(state.numPartitions, count)

case _ => 0

}

val partitions = bornDisklessPartitions(topicId, count).filter(_ >= firstPartition)

val retry = errorCode != Errors.NONE.code()

val imagePartitions = Option(metadataCache.currentImage().topics().getTopic(topicId))

.map(_.partitions())

.getOrElse(util.Collections.emptyMap())

if (retry && count < imagePartitions.size()) {

Seq.empty

} else {

val firstPartition =

if (retry) 0

else priorTopicStates.get(topicName) match {

case Some(state) if state.topicId == topicId => math.min(state.numPartitions, count)

case _ => 0

}

val partitions = bornDisklessPartitions(imagePartitions, count)....

jeqo · 2026-08-17T18:33:16Z

+    val endPoint = listOffsetEndPointWithPlaceholderEpoch(offset = 10L, seal = 150000L, disklessLeaderEpoch = 5)
+    assertEquals(new OffsetAndEpoch(150000L, 5), endPoint.fetchLatestOffset(topicPartition, 3))
+  }
+


for completeness, adding the negative

Suggested change

@Test

def testFetchLatestOffsetSwitchPendingSealIsUnchanged(): Unit = {

val endPoint = listOffsetEndPointWithPlaceholderEpoch(

offset = 0L,

seal = PartitionRegistration.CLASSIC_TO_DISKLESS_SWITCH_PENDING,

disklessLeaderEpoch = PartitionRegistration.NO_DISKLESS_LEADER_EPOCH)

assertEquals(new OffsetAndEpoch(0L, 0), endPoint.fetchLatestOffset(topicPartition, 3))

}

@Test

def testFetchLatestOffsetBornDisklessSealIsUnchanged(): Unit = {

val endPoint = listOffsetEndPointWithPlaceholderEpoch(

offset = 0L,

seal = PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET,

disklessLeaderEpoch = PartitionRegistration.NO_DISKLESS_LEADER_EPOCH)

assertEquals(new OffsetAndEpoch(0L, 0), endPoint.fetchLatestOffset(topicPartition, 3))

}

@Test

def testFetchLatestOffsetExactlyAtSealIsUnchanged(): Unit = {

val endPoint = listOffsetEndPointWithPlaceholderEpoch(offset = 150000L, seal = 150000L, disklessLeaderEpoch = 5)

assertEquals(new OffsetAndEpoch(150000L, 5), endPoint.fetchLatestOffset(topicPartition, 3))

}

jeqo · 2026-08-17T18:40:16Z

+          .filter { case (_, res) =>
+            val code = res.errorCode()
+            code == Errors.NONE.code() ||
+              code == Errors.TOPIC_ALREADY_EXISTS.code() ||


does this error actually need to be covered? I think it comes from topic creation which is not reachable when creating partitions

viktorsomogyi force-pushed the svv/fix-create-partitions-missing-logs-rows branch from 370f426 to 7e6c736 Compare August 14, 2026 13:43

viktorsomogyi requested a lite review from Copilot August 14, 2026 13:44

Copilot started reviewing on behalf of viktorsomogyi August 14, 2026 13:45 View session

Copilot AI reviewed Aug 14, 2026

View reviewed changes

Comment thread ...ge/inkless/src/main/java/io/aiven/inkless/control_plane/CreateTopicAndPartitionsRequest.java

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

viktorsomogyi force-pushed the svv/fix-create-partitions-missing-logs-rows branch from 7e6c736 to 9f28516 Compare August 14, 2026 13:54

viktorsomogyi changed the title ~~fix(inkless): repair missing control-plane rows on CreatePartitions retry~~ fix(inkless): repair missing control-plane rows on CreatePartitions retry [KC-420] Aug 14, 2026

viktorsomogyi marked this pull request as ready for review August 14, 2026 14:36

viktorsomogyi requested a review from jeqo August 14, 2026 14:36

jeqo reviewed Aug 17, 2026

View reviewed changes

Navigation Menu

Uh oh!

fix(inkless): repair missing control-plane rows on CreatePartitions retry [KC-420] - #752

fix(inkless): repair missing control-plane rows on CreatePartitions retry [KC-420]#752
viktorsomogyi wants to merge 1 commit into
mainfrom
svv/fix-create-partitions-missing-logs-rows

viktorsomogyi commented Aug 14, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

jeqo left a comment

Uh oh!

jeqo Aug 17, 2026

Uh oh!

jeqo Aug 17, 2026

Uh oh!

jeqo Aug 17, 2026

Uh oh!

jeqo Aug 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

+  @Test
+  def testCreateDisklessPartitionsDecreaseRetryDoesNotInsert(): Unit = {
+    // topic image already has 4 partitions; a decrease to 2 is rejected by KRaft with
+    // INVALID_PARTITIONS and must NOT write control-plane rows.
+    val topic = new CreatePartitionsTopic().setName(topicName).setAssignments(null).setCount(2)
+    val result = new CreatePartitionsTopicResult().setName(topicName).setErrorCode(INVALID_PARTITIONS.code())
+    // ... arrange metadataCache image with 4 partitions for topicId (as the other retry tests do) ...
+    // invoke createDisklessPartitions(...)
+    verify(controlPlane, never()).createTopicAndPartitions(any())
+  }

+@Test
+def testFetchLatestOffsetSwitchPendingSealIsUnchanged(): Unit = {
+  val endPoint = listOffsetEndPointWithPlaceholderEpoch(
+    offset = 0L,
+    seal = PartitionRegistration.CLASSIC_TO_DISKLESS_SWITCH_PENDING,
+    disklessLeaderEpoch = PartitionRegistration.NO_DISKLESS_LEADER_EPOCH)
+  assertEquals(new OffsetAndEpoch(0L, 0), endPoint.fetchLatestOffset(topicPartition, 3))
+}
+@Test
+def testFetchLatestOffsetBornDisklessSealIsUnchanged(): Unit = {
+  val endPoint = listOffsetEndPointWithPlaceholderEpoch(
+    offset = 0L,
+    seal = PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET,
+    disklessLeaderEpoch = PartitionRegistration.NO_DISKLESS_LEADER_EPOCH)
+  assertEquals(new OffsetAndEpoch(0L, 0), endPoint.fetchLatestOffset(topicPartition, 3))
+}
+@Test
+def testFetchLatestOffsetExactlyAtSealIsUnchanged(): Unit = {
+  val endPoint = listOffsetEndPointWithPlaceholderEpoch(offset = 150000L, seal = 150000L, disklessLeaderEpoch = 5)
+  assertEquals(new OffsetAndEpoch(150000L, 5), endPoint.fetchLatestOffset(topicPartition, 3))
+}

Uh oh!

Conversation

viktorsomogyi commented Aug 14, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

jeqo left a comment

Choose a reason for hiding this comment

Uh oh!

jeqo Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

jeqo Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

jeqo Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

jeqo Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants